package com.zz.server.handle;

import com.zz.common.config.Constants;
import com.zz.common.datasource.DataSource;
import com.zz.common.datasource.DataSourceFactory;
import com.zz.common.entity.ClientConfig;
import com.zz.common.entity.ProtocolInfo;
import com.zz.common.enums.MessageType;
import com.zz.common.enums.ProxyType;
import com.zz.common.enums.Status;
import com.zz.common.function.Dispatch;
import com.zz.common.handle.executor.MessageExecutor;
import com.zz.common.handle.executor.MessageExecutorFactory;
import com.zz.common.utils.ObjectUtil;
import com.zz.common.utils.StringUtil;
import com.zz.common.coder.ByteArrayCodec;
import com.zz.common.coder.HttpEncoder;
import com.zz.server.config.ChannelCache;
import com.zz.server.handle.web.page.StaticDispatcher;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.internal.EmptyArrays;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 *  http
 *
 * @author zz
 */
@Slf4j
public class HttpChannelHandle extends SimpleChannelInboundHandler<FullHttpRequest> {

    private final DataSource dataSource = DataSourceFactory.getTunnelSource();
    private final Queue<FullHttpRequest> messageQueue = new LinkedBlockingQueue<>();

    private final AtomicBoolean connected = new AtomicBoolean(false);

    private final HttpEncoder.RequestEncoder requestEncoder = new HttpEncoder.RequestEncoder();

    private String domain;


    @Override
    public void exceptionCaught(ChannelHandlerContext context, Throwable cause) throws Exception {
        context.channel().close();
        log.error("http 通道异常！error：{}", cause.toString());
    }

    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        Channel proxyChannel = ctx.channel().attr(Constants.Attributes.LOCAL_CHANNEL).get();
        if (ObjectUtil.isNotNull(proxyChannel)) {
            proxyChannel.config().setAutoRead(ctx.channel().isWritable());
        }
        super.channelWritabilityChanged(ctx);
    }

    /**
     * 连接中断
     */
    @Override
    public void channelInactive(ChannelHandlerContext context) throws Exception {
        ObjectUtil.setReadable(context);
        // 释放队列中消息
        messageQueueHandle(null);
        if (ObjectUtil.isEmpty(domain)) {
            return;
        }
        Channel masterServerChannel = ChannelCache.getChannelByDomain(domain);
        if (ObjectUtil.isEmpty(masterServerChannel)) {
            return;
        }
        // 清空主通道中关联的当前channel
        Map<String, Channel> channelCache = masterServerChannel.attr(Constants.Attributes.CHANNEL_CACHE).get();
        if (ObjectUtil.isNotEmpty(channelCache)) {
            channelCache.remove(context.channel().id().asLongText());
        }
        // 清空从通道中的关联的当前channel相关信息
        Channel salveServerChannel = context.channel().attr(Constants.Attributes.SERVER_CHANNEL).getAndSet(Constants.Objects.CHANNEL_NULL);
        if (ObjectUtil.isEmpty(salveServerChannel)) {
            return;
        }
        salveServerChannel.attr(Constants.Attributes.LOCAL_CHANNEL).set(Constants.Objects.CHANNEL_NULL);
        ObjectUtil.setReadable(salveServerChannel);
        // 发送断开连接请求
        transmit(masterServerChannel, context, domain, MessageType.TYPE_DISCONNECT, EmptyArrays.EMPTY_BYTES);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext context, Object event) throws Exception {
        // 当连接建立后手动触发该事件，事件类型为HTTP
        if (!ProxyType.HTTP.equals(event)) {
            context.fireUserEventTriggered(event);
            return;
        }
        Channel salveServerChannel = context.channel().attr(Constants.Attributes.SERVER_CHANNEL).get();
        if (ObjectUtil.isEmpty(salveServerChannel)) {
            log.error("salveServerChannel is null");
            return;
        }
        // 替换http解析处理器为字节处理器
        context.pipeline().replace(HttpServerCodec.class, ByteArrayCodec.class.getName(), new ByteArrayCodec());
        connected.set(true);
        // 并发环境下防止队列消息无法消费
        synchronized (connected) {
            if (messageQueue.isEmpty()) {
                return;
            }
            messageQueueHandle(request -> httpRequestHandle(context, salveServerChannel, request));
        }
        ObjectUtil.setReadable(context);
    }

    /**
     * 处理消息队列数据并释放http对象
     * @param dispatch
     * @throws Exception
     */
    private void messageQueueHandle(Dispatch<FullHttpRequest> dispatch) throws Exception {
        FullHttpRequest request;
        while (ObjectUtil.isNotNull((request = messageQueue.poll()))) {
            if (ObjectUtil.isNotNull(dispatch)) {
                dispatch.accept(request);
            }
            ReferenceCountUtil.release(request);
        }
    }

    /**
     * 当第一次请求时，通过http信息解析域名并缓存且替换http解析处理器为字节处理器后续请求直接转发
     * @param context
     * @param obj
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext context, Object obj) throws Exception {
        if (obj instanceof FullHttpRequest) {
            super.channelRead(context, obj);
        } else if (obj instanceof byte[]) {
            byte[] message = (byte[]) obj;
            Channel salveServerChannel = context.channel().attr(Constants.Attributes.SERVER_CHANNEL).get();
            if (ObjectUtil.isEmptys(salveServerChannel, message)) {
                context.close();
                return;
            }
            transmit(salveServerChannel, context, domain, MessageType.TYPE_TRANSFER, message);
        } else {
            context.fireChannelRead(obj);
        }
    }


    @Override
    protected void channelRead0(ChannelHandlerContext context, FullHttpRequest httpRequest) throws Exception {
        // 解析域名
        domain = Optional.ofNullable(domain).orElse(Constants.Objects.COLON.split(httpRequest.headers().get(HttpHeaderNames.HOST))[0]);
        if (!StringUtil.hasText(domain) || !isOpen(domain)) {
            StaticDispatcher.dispatch(httpRequest, context);
            return;
        }
        Channel masterChannel = ChannelCache.getChannelByDomain(domain);
        if (ObjectUtil.isEmpty(masterChannel) || !masterChannel.isOpen()) {
            StaticDispatcher.dispatch(httpRequest, context);
            return;
        }
        // 检查是否建立连接，如无发送连接建立请求，并将消息放入队列中
        if (!connected.get()) {
            synchronized (connected) {
                if (!connected.get()) {
                    // 设置当前channel不可自动读
                    ObjectUtil.setNotReadable(context);
                    masterChannel.attr(Constants.Attributes.CHANNEL_CACHE).get().put(context.channel().id().asLongText(), context.channel());
                    transmit(masterChannel, context, domain, MessageType.TYPE_CONNECT, EmptyArrays.EMPTY_BYTES);
                    messageQueue.offer(httpRequest.retain());
                    return;
                }
            }
        }
        // 获取当前连接的业务处理 channel
        Channel slaveSeverChannel = context.channel().attr(Constants.Attributes.SERVER_CHANNEL).get();
        if (ObjectUtil.isEmpty(slaveSeverChannel) || !slaveSeverChannel.isOpen()) {
            StaticDispatcher.dispatch(httpRequest, context);
            return;
        }
        // 编码http请求 发送客户端
        httpRequestHandle(context, slaveSeverChannel, httpRequest);
    }

    private void httpRequestHandle(ChannelHandlerContext context, Channel slaveSeverChannel, FullHttpRequest httpRequest) throws Exception {
        List<Object> requestEncode = new ArrayList<>();
        requestEncoder.encode(context, httpRequest, requestEncode);
        requestEncode.forEach(obj -> {
            if (!(obj instanceof ByteBuf)) {
                return;
            }
            transmit(slaveSeverChannel, context, domain, MessageType.TYPE_TRANSFER, ObjectUtil.byteProcess((ByteBuf) obj));
        });
    }



    /**
     * 当前用户配置是否开启代理
     * @param domain
     * @return
     */
    private boolean isOpen(String domain) {
        ClientConfig config = dataSource.getClientConfigIdByDomain(domain);
        return Objects.nonNull(config) && config.getStatus() == Status.OPEN;
    }


    /**
     * 数据转发
     * @param salveChannel
     * @param context
     * @param domain
     * @param messageType
     * @param data
     */
    private void transmit(Channel salveChannel, ChannelHandlerContext context, String domain, MessageType messageType, byte[] data) {
        ProtocolInfo protocolInfo = ProtocolInfo.builder()
                .serverChannel(salveChannel).context(context).domain(domain).type(messageType).data(data).build();

        MessageExecutor executor = MessageExecutorFactory.getMessageExecutor(messageType);
        if (Objects.isNull(executor)) {
            log.info("错误！找不到对应的消息处理器！messageType:{}", messageType);
            return;
        }
        executor.execute(protocolInfo);
    }




}
